package org.idea.qiyu.framework.canal.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.idea.qiyu.framework.canal.annotation.CanalListener;
import org.idea.qiyu.framework.canal.consumer.CanalMessageConsumer;
import org.idea.qiyu.framework.canal.handle.AbstractCanalHandler;
import org.idea.qiyu.framework.canal.handle.CanalMessageQueueHandler;
import org.idea.qiyu.framework.canal.producer.CanalMessageProducer;
import org.idea.qiyu.framework.canal.properties.CanalInstanceProperties;
import org.idea.qiyu.framework.canal.properties.CanalProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * canal相关的自动化配置类
 *
 * @Author linhao
 * @Date created in 4:55 下午 2021/6/19
 */
@Configuration
@EnableConfigurationProperties(CanalProperties.class)
public class CanalConfiguration implements ApplicationContextAware {

    private static final Logger LOGGER = LoggerFactory.getLogger(CanalConfiguration.class);

    @Resource
    private CanalProperties canalProperties;
    @Resource
    private CanalMessageQueueHandler canalMessageQueueHandler;

    private static Map<String, CanalConnector> canalConnectorMap = new ConcurrentHashMap<>();

    /**
     * 先setApplicationContext 再触发afterPropertiesSet
     *
     * @param applicationContext
     * @throws BeansException
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        CanalMessageConsumer canalMessageConsumer = new CanalMessageConsumer(canalMessageQueueHandler);
        Map<String, CanalInstanceProperties> canalInstancePropertiesMap = canalProperties.getInstances();
        for (String instanceName : canalInstancePropertiesMap.keySet()) {
            CanalInstanceProperties canalInstanceProperties = canalInstancePropertiesMap.get(instanceName);
            CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalInstanceProperties.getAddress(), canalInstanceProperties.getPort()),
                    instanceName, canalInstanceProperties.getAccount(), canalInstanceProperties.getPassword());
            canalConnector.connect();
            canalConnector.subscribe(canalInstanceProperties.getFilter());
            canalConnectorMap.put(instanceName, canalConnector);

            LOGGER.info("[CanalAutoConfiguration] init canalConnector success! canalConnector is {}", canalConnector);
        }

        Map<String, AbstractCanalHandler> canalHandlerMap = applicationContext.getBeansOfType(AbstractCanalHandler.class);
        Set<CanalConnector> canalConnectorSet = new HashSet<>();
        for (String listenerName : canalHandlerMap.keySet()) {
            AbstractCanalHandler abstractCanalHandler = canalHandlerMap.get(listenerName);
            if(abstractCanalHandler==null){
                continue;
            }
            CanalListener canalListener = abstractCanalHandler.getClass().getAnnotation(CanalListener.class);
            abstractCanalHandler.setInstanceName(canalListener.destination());
            abstractCanalHandler.setTableName(canalListener.table());
            canalMessageConsumer.addConsume(abstractCanalHandler);
            CanalConnector canalConnector = canalConnectorMap.get(canalListener.destination());
            canalConnectorSet.add(canalConnector);
            LOGGER.info("[canalListener] " + listenerName + " is started!");
        }
        Thread canalMessageProducerThread = new Thread(new CanalMessageProducer(canalConnectorSet, canalMessageQueueHandler));
        canalMessageProducerThread.start();
        Thread canalMessageConsumerThread = new Thread(canalMessageConsumer);
        canalMessageConsumerThread.start();
    }
}
